package com.hzw.saas.common.storage.util;

import java.io.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.io.IoUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class FileStreamUtil {

    static int BUF_SIZE = 8192;

    /**
     * 定长转发
     */
    public static void readAndWrite(InputStream ins, OutputStream ous, long length, int BUF_SIZE, boolean closeOus) throws Exception {
        byte[] buff = new byte[BUF_SIZE];

        int remain = (int) length;
        int avalible = BUF_SIZE;

        try {
            while (remain > 0) {
                if (log.isTraceEnabled()) {
                    log.trace("remain:" + remain);
                }

                if (remain < BUF_SIZE) {
                    avalible = remain;
                }
                if (log.isTraceEnabled()) {
                    log.trace("avalible:" + avalible);
                }

                int len = ins.read(buff, 0, avalible);
                if (log.isTraceEnabled()) {
                    log.trace("readlen:" + len);
                }

                if (len < 0) {
                    log.warn("read len < 0, filedata is missing");
                    break;
                }

                ous.write(buff, 0, len);
                ous.flush();
                remain -= len;
            }

        } catch (Exception e) {
            log.error("receive file data error", e);
            throw e;
        } finally {
            if (null != ous && closeOus) {
                ous.close();
            }
        }
    }

    /**
     * 未知长度转发（读取到-1为止）
     */
    public static long readAndWrite(InputStream ins, OutputStream ous, int BUF_SIZE, boolean closeOus) throws IOException {
        long length = 0;
        byte[] buf = new byte[BUF_SIZE];
        int size = 0;
        try {
            while ((size = ins.read(buf)) != -1) {
                ous.write(buf, 0, size);
                length += size;
            }
        } finally {
            if (null != ous && closeOus) {
                ous.close();
            }
        }
        return length;
    }

    public static void pipeStream(InputStream ins, OutputStream ous) throws IOException {
        ExecutorService executor = new ThreadPoolExecutor(1, 3,
            60L, TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>());
        // read ins write to pous
        final PipedOutputStream pous = new PipedOutputStream();
        executor.submit(() -> {
            try {
                readAndWrite(ins, pous, BUF_SIZE, true);
            } catch (Exception e) {
                throw ExceptionUtil.wrapRuntime(e);
            } finally {
                IoUtil.close(ins);
            }
        });
        // read pins write to ous
        final PipedInputStream pins = new PipedInputStream(pous, BUF_SIZE);
        try {
            readAndWrite(pins, ous, BUF_SIZE, false);
        } catch (Exception e) {
            throw ExceptionUtil.wrapRuntime(e);
        } finally {
            IoUtil.close(pins);
            IoUtil.close(pous);
        }
    }

}
